package com.atguigu.flink.datastream.transform;

import com.atguigu.flink.pojo.AppEvent;
import com.atguigu.flink.pojo.ThirdPartyEvent;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author WEIYUNHUI
 * @date 2023/6/14 10:16
 *
 * 对账案例:
 *    我们可以实现一个实时对账的需求，也就是app的支付操作和第三方的支付操作的一个双流Join。
 *    目前只关心对账成功的数据。 不限制两方数据的延迟(死等) 
 */
public class Flink11_BillCheck {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        //基于端口模拟数据
        //AppEvent 
        //order-1,create,1000
        //order-1,pay,2000
        SingleOutputStreamOperator<AppEvent> appDs = env.socketTextStream("hadoop102", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new AppEvent(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
                        }
                ).filter(
                        event -> "pay".equals(event.getEventType())
                );
        
        // ThirdPartyEvent
        // order-1,pay,wechat,2000
        SingleOutputStreamOperator<ThirdPartyEvent> thirdDs = env.socketTextStream("hadoop102", 9999)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new ThirdPartyEvent(fields[0].trim(), fields[1].trim(), fields[2].trim(), Long.valueOf(fields[3].trim()));
                        }
                );
        
        //connect
        SingleOutputStreamOperator<String> processDs = appDs.connect(thirdDs)
                .process(
                        new CoProcessFunction<AppEvent, ThirdPartyEvent, String>() {

                            //使用集合维护早到的数据
                            //自己维护状态
                            private Map<String, AppEvent> appMap = new HashMap<>();
                            private Map<String, ThirdPartyEvent> thirdMap = new HashMap<>();

                            /**
                             * 处理AppEvent
                             */
                            @Override
                            public void processElement1(AppEvent value, CoProcessFunction<AppEvent, ThirdPartyEvent, String>.Context ctx, Collector<String> out) throws Exception {
                                //先到对方法的状态中看看是否有对应的数据
                                if (thirdMap.containsKey(value.getOrderId())) {
                                    out.collect(value.getOrderId() + "对账成功!!!");
                                    //从对方的状态中清除数据
                                    thirdMap.remove(value.getOrderId());
                                } else {
                                    // appEvent到了， thirdPartyEvent没到
                                    appMap.put(value.getOrderId(), value);
                                }
                            }

                            /**
                             * 处理ThirdPartyEvent
                             */
                            @Override
                            public void processElement2(ThirdPartyEvent value, CoProcessFunction<AppEvent, ThirdPartyEvent, String>.Context ctx, Collector<String> out) throws Exception {
                                //先到对方法的状态中看看是否有对应的数据
                                if (appMap.containsKey(value.getOrderId())) {
                                    out.collect(value.getOrderId() + "对账成功!!!");
                                    //从对方的状态中清除数据
                                    appMap.remove(value.getOrderId());
                                } else {
                                    // thirdPartyEvent到了， appEvent没到
                                    thirdMap.put(value.getOrderId(), value);
                                }
                            }

                        }
                );

        processDs.print();
       
        /*
        appDs.keyBy(
               AppEvent::getOrderId 
        ).connect(
               thirdDs.keyBy(
                       ThirdPartyEvent::getOrderId
               ) 
        )
         */
                

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
